package com.gedi.springboot_kafka.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author:RenPu
 * @Date: 2020/3/19 21:02
 * @Version: 1.0
 * @description:
 */
//当前的类，自动注册为spring容器中的bean
@Component
public class ProducerDemo {


    @Autowired
    private KafkaTemplate<Integer,String> kafkaTemplate;

    //原子性操作的整数类，线程属于安全的
    private AtomicInteger atomicInteger=new AtomicInteger();



    //每五秒发送一次
    @Scheduled(cron = "0/5 * * * * ?")
    public void send(){

        //获取递增长的整数
        int num = atomicInteger.getAndIncrement();

        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send("t4", num, "Hello World" + num);


           //匿名内部类的写法
//        future.addCallback(new SuccessCallback<SendResult<Integer, String>>() {
//            @Override
//            public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
//
//            }
//        }, new FailureCallback() {
//            @Override
//            public void onFailure(Throwable throwable) {
//
//            }
//        });


        //lamda写法或者函数编程写法
        future.addCallback((t->{

            System.out.println("发送成功"+t);

        }),(e->{
            System.out.println("发送失败"+e.getMessage());

        }));

    }
}
